package com.rtf.framework.es.service;

import com.rtf.framework.common.json.JsonMapper;
import com.rtf.framework.common.util.UtilCollection;
import com.rtf.framework.common.util.UtilMap;
import com.rtf.framework.common.util.UtilString;
import com.rtf.framework.es.common.*;
import com.rtf.framework.es.exception.*;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.lucene.search.TotalHits;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.*;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.client.indices.GetIndexResponse;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.script.Script;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.metrics.ValueCount;
import org.elasticsearch.search.aggregations.metrics.ValueCountAggregationBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.elasticsearch.search.sort.SortBuilder;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import java.io.IOException;
import java.util.*;

/**
 * ElasticSearch操作服务
 * @Author : lwq
 * @Date : 2021-06-02
 * @Modified By
 */
@Slf4j
@Service
public class AppElasticSearchServiceImpl<T extends AppEsEntity> implements AppElasticSearchService<T>{
    private static final int STATUS_STATUS = 200;

    private RestHighLevelClient restHighLevelClient;

    public AppElasticSearchServiceImpl(RestHighLevelClient restHighLevelClient) {
        this.restHighLevelClient = restHighLevelClient;
    }

    /**
     * 创建索引，新版ES插入数据时自动创建
     *
     * @param index      索引
     * @param shardNum   主分片数量
     * @param replicaNum 副本数量
     * @return
     */
    @Override
    public boolean createIndex(String index, Map<String,?> mapping, Integer shardNum, Integer replicaNum) {
        try {
            CreateIndexRequest createIndexRequest = new CreateIndexRequest(index);
            createIndexRequest.mapping(mapping);
            //设置分片和副本
            createIndexRequest.settings(Settings.builder()
                    .put("index.number_of_shards", shardNum)
                    .put("index.number_of_replicas", replicaNum)
            );
            CreateIndexResponse response = restHighLevelClient.indices().create(createIndexRequest, RequestOptions.DEFAULT);
            return response.isAcknowledged();
        } catch (Exception e) {
            log.error("创建索引失败：", e);
            throw new AppEsCreateException(e);
        }
    }

    /**
     * 获取索引
     *
     * @throws IOException
     */
    @Override
    public GetIndexResponse getIndex(String index) {
        try {
            //创建获取索引请求
            GetIndexRequest request = new GetIndexRequest(index);
            //执行获取索引请求判断是否有这个索引
            GetIndexResponse response = restHighLevelClient.indices().get(request, RequestOptions.DEFAULT);
            return response;
        } catch (Exception e) {
            log.error("获取索引信息失败：", e);
            throw new AppEsQueryException(e);
        }
    }

    /**
     * 判断是否拥有某个索引
     *
     */
    @Override
    public Boolean existsIndex(String index) {
        try {
            //创建获取索引请求
            GetIndexRequest getRequest = new GetIndexRequest(index);
            //执行获取索引请求判断是否有这个索引
            return restHighLevelClient.indices().exists(getRequest, RequestOptions.DEFAULT);
        } catch (Exception e) {
            log.error("判断es索引是否存在失败", e);
            throw new AppEsQueryException(e);
        }
    }

    /**
     * 删除索引
     */
    @Override
    public boolean deleteIndex(String index) {
        try {
            DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(index);
            AcknowledgedResponse response = restHighLevelClient.indices().delete(deleteIndexRequest, RequestOptions.DEFAULT);
            return response.isAcknowledged();
        } catch (Exception e) {
            log.error("删除索引失败：", e);
            throw new AppEsDeleteException(e);
        }
    }

    /**
     * 更新单条数据
     */
    @Override
    public void updateDocument(String index, T esEntity) {
        Assert.isTrue(esEntity != null && UtilString.isNotBlank(esEntity.getId()), "id不能为null");
        try {
            UpdateRequest updateRequest = new UpdateRequest(index, String.valueOf(esEntity.getId()));
            updateRequest.doc(JsonMapper.toJSONString(esEntity), XContentType.JSON);
            updateRequest.id(String.valueOf(esEntity.getId()));
            UpdateResponse updateResponse = restHighLevelClient.update(updateRequest, RequestOptions.DEFAULT);
            if (updateResponse.status().getStatus() != STATUS_STATUS) {
                log.error("修改数据失败：{}", updateResponse.toString());
                throw new AppEsUpdateException(updateResponse.toString());
            }
        } catch (Exception e) {
            log.error("修改数据失败：", e);
            throw new AppEsUpdateException(e);
        }
    }

    /**
     * 更新单条数据
     */
    @Override
    public void updateDocument(String index, Map<String, Object> params) {
        Assert.isTrue(params != null && UtilMap.getString(params, "id") != null, "id不能为null");
        try {
            UpdateRequest updateRequest = new UpdateRequest(index, UtilMap.getString(params, "id"));
            updateRequest.doc(params);
            UpdateResponse updateResponse = restHighLevelClient.update(updateRequest, RequestOptions.DEFAULT);
            if (updateResponse.status().getStatus() != STATUS_STATUS) {
                log.error("es更新数据失败：{}", updateResponse.toString());
                throw new AppEsUpdateException("es更新数据失败");
            }
        } catch (Exception e) {
            log.error("更新失败", e);
            throw new AppEsUpdateException(e);
        }
    }

    /**
     * 批量更新
     *
     * @param index
     * @param list
     * @return
     * @throws IOException
     */
    @Override
    public void updateBatch(String index, List<Map<String, Object>> list) {
        Assert.isTrue(UtilCollection.isNotEmpty(list), "数据不能为空");
        try {
            BulkRequest bulkRequest = new BulkRequest();
            String id;
            for (Map<String, Object> info : list) {
                id = UtilMap.getString(info, "id");
                Assert.isTrue(UtilString.isNotBlank(id), "id不能为null");
                UpdateRequest request = new UpdateRequest(index,  id)
                        .doc(JsonMapper.toJSONString(info), XContentType.JSON);
                bulkRequest.add(request);
            }
            BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (bulk.hasFailures()) {
                log.error("批量更新失败：{}", bulk.buildFailureMessage());
                throw new AppEsUpdateException(bulk.buildFailureMessage());
            }
        } catch (Exception e) {
            log.error("批量更新失败：", e);
            throw new AppEsUpdateException(e);
        }
    }

    /**
     * 批量更新数据
     *
     * @param index
     * @param list
     * @return
     */
    @Override
    public void updateBatchDocument(String index, List<T> list) {
        Assert.isTrue(UtilCollection.isNotEmpty(list), "数据不能为空");
        try {
            BulkRequest bulkRequest = new BulkRequest();
            for (T t : list) {
                Assert.isTrue(t != null && UtilString.isNotBlank(t.getId()), "id不能为null");
                UpdateRequest request = new UpdateRequest(index, String.valueOf(t.getId())).doc(JsonMapper.toJSONString(t), XContentType.JSON);
                bulkRequest.add(request);
            }
            BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (bulk.hasFailures()) {
                log.error("批量更新失败：{}", bulk.buildFailureMessage());
                throw new AppEsUpdateException(bulk.buildFailureMessage());
            }
        } catch (Exception e) {
            log.error("批量更新失败：", e);
            throw new AppEsUpdateException(e);
        }
    }

    /**
     * 批量更新
     *
     * @param index
     * @param list
     * @return BulkResponse
     * @throws IOException
     */
    @Override
    public void updateBatchByScript(String index, List<AppEsScriptInfo> list) {
        Assert.isTrue(UtilCollection.isNotEmpty(list), "数据不能为空");
        try {
            BulkRequest bulkRequest = new BulkRequest();
            for (AppEsScriptInfo info : list) {
                Assert.isTrue(info != null && UtilString.isNotBlank(info.getId()), "id不能为null");
                UpdateRequest request = new UpdateRequest(index, info.getId()).script(info.getScript());
                bulkRequest.add(request);
            }
            BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (bulk.hasFailures()) {
                log.error("批量更新失败：{}", bulk.buildFailureMessage());
                throw new AppEsUpdateException(bulk.buildFailureMessage());
            }
        } catch (Exception e) {
            log.error("批量更新失败：", e);
            throw new AppEsUpdateException(e);
        }
    }

    /**
     * 批量更新
     *
     * @param index
     * @param queryBuilder  查询条件
     * @param batchSize     批量更改大小
     * @throws IOException
     */
    @Override
    public void updateByQuery(String index, QueryBuilder queryBuilder, Integer batchSize, Map<String, Object> document) {
        Assert.isTrue(UtilMap.isNotEmpty(document), "更新字段不能为空");
        try {
            UpdateByQueryRequest request = new UpdateByQueryRequest(index);
            request.setQuery(queryBuilder);
            request.setBatchSize(batchSize == null || batchSize < 1 ? AppEsConstants.DEFAULT_UPDATE_BATCH_SIZE : batchSize);
            StringBuilder sb = new StringBuilder(STATUS_STATUS);
            if (!UtilMap.isEmpty(document)) {
                Iterator<Map.Entry<String, Object>> iterator = document.entrySet().iterator();
                Map.Entry<String, Object> next;
                while (iterator.hasNext()) {
                    next = iterator.next();
                    sb.append("ctx._source.").append(next.getKey()).append("=params.").append(next.getKey()).append(";");
                }
                request.setScript(new Script(Script.DEFAULT_SCRIPT_TYPE, Script.DEFAULT_SCRIPT_LANG, sb.toString(), document));
            }
            BulkByScrollResponse response = restHighLevelClient.updateByQuery(request, RequestOptions.DEFAULT);
            if (response.getBulkFailures().size() > 0) {
                for (BulkItemResponse.Failure bulkFailure : response.getBulkFailures()) {
                    log.error("批量更新失败：{}", bulkFailure.getMessage());
                }
                throw new AppEsUpdateException("更新失败");
            }
        } catch (Exception e) {
            log.error("批量更新失败：", e);
            throw new AppEsUpdateException(e);
        }
    }

    /**
     * 批量更新
     *
     * @param index
     * @param queryBuilder 查询条件
     * @param batchSize    批量更新大小
     * @throws IOException
     */
    @Override
    public void updateByQuery(String index, QueryBuilder queryBuilder, Integer batchSize, Script script) {
        try {
            UpdateByQueryRequest request = new UpdateByQueryRequest(index);
            request.setQuery(queryBuilder);
            request.setBatchSize(batchSize == null || batchSize < 1 ? AppEsConstants.DEFAULT_UPDATE_BATCH_SIZE : batchSize);
            request.setScript(script);
            BulkByScrollResponse response = restHighLevelClient.updateByQuery(request, RequestOptions.DEFAULT);
            if (response.getBulkFailures().size() > 0) {
                for (BulkItemResponse.Failure bulkFailure : response.getBulkFailures()) {
                    log.error("批量更新失败：{}", bulkFailure.getMessage());
                }
                throw new AppEsUpdateException("更新失败");
            }
        } catch (Exception e) {
            log.error("查询更新失败：", e);
            throw new AppEsUpdateException(e);
        }
    }

    /**
     * 保存信息
     *
     * @param index      索引名称
     * @param esEntities
     * @return IndexResponse
     * @throws Exception
     * @author lwq
     */
    @Override
    public void insertDocument(String index, T esEntities) {
        Assert.isTrue(esEntities != null && esEntities.getId() != null, "保存信息不能为空");
        try {
            IndexRequest indexRequest = new IndexRequest(index);
            indexRequest.id(String.valueOf(esEntities.getId()));
            indexRequest.source(JsonMapper.toJSONString(esEntities), XContentType.JSON);
            IndexResponse response = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
            if (!response.status().equals(RestStatus.CREATED)) {
                log.error("保存信息失败：{}", response.toString());
                throw new AppEsCreateException(response.toString());
            }
        } catch (Exception e) {
            log.error("", e);
            throw new AppEsCreateException(e);
        }
    }


    /**
     * 批量插入数据
     */
    @Override
    public void insertBatchDocument(String index, List<T> esEntities) {
        Assert.isTrue(UtilCollection.isNotEmpty(esEntities), "数据不能为空");
        try {
            BulkRequest bulkRequest = new BulkRequest();
            for (AppEsEntity umsEsEntity : esEntities) {
                IndexRequest indexRequest = new IndexRequest(index);
                indexRequest.id(umsEsEntity.getId());
                indexRequest.source(JsonMapper.toJSONString(umsEsEntity), XContentType.JSON);
                bulkRequest.add(indexRequest);
            }
            BulkResponse response = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (response.hasFailures()) {
                log.error("批量插入数据失败：{}", response.buildFailureMessage());
                throw new AppEsCreateException(response.buildFailureMessage());
            }
        } catch (Exception e) {
            log.error("批量插入数据失败", e);
            throw new AppEsCreateException(e);
        }
    }

    /**
     * 通过ID获取数据
     *
     * @param index         索引，类似数据库
     * @param id            数据ID
     * @param includeFields 需要显示的字段
     * @param excludeFields 不需要显示的字段
     * @return Map
     * @author lwq
     */
    @Override
    public Map<String, Object> getDataById(String index, String id, String[] includeFields, String[] excludeFields) {
        Assert.isTrue(UtilString.isNotEmpty(id), "id不能为空");
        try {
            GetRequest request = new GetRequest(index, id);
            if (includeFields == null) {
                includeFields = Strings.EMPTY_ARRAY;
            }
            if (excludeFields == null) {
                excludeFields = Strings.EMPTY_ARRAY;
            }
            //只查询特定字段。如果需要查询所有字段则不设置该项。
            request.fetchSourceContext(new FetchSourceContext(true, includeFields, excludeFields));
            GetResponse response = restHighLevelClient.get(request, RequestOptions.DEFAULT);
            return response.getSource();
        } catch (Exception e) {
            log.error("根据id es查询失败", e);
            throw new AppEsQueryException(e);
        }
    }

    /**
     * 通过ID获取数据
     *
     * @param index 索引，类似数据库
     * @param id    数据ID
     * @return
     * @author lwq
     */
    @Override
    public Map<String, Object> getDataById(String index, String id) {
        return getDataById(index, id, null, null);
    }

    /**
     * 通过ID获取数据
     *
     * @param index         索引，类似数据库
     * @param id            数据ID
     * @param includeFields 需要显示的字段
     * @return
     */
    @Override
    public Map<String, Object> getDataByIdIncludeFields(String index, String id, String[] includeFields) {
        return getDataById(index, id, includeFields, null);

    }

    /**
     * 通过ID获取数据
     *
     * @param index         索引，类似数据库
     * @param id            数据ID
     * @param excludeFields 不需要显示的字段
     * @return
     */
    @Override
    public Map<String, Object> getDataByIdExcludeFields(String index, String id, String[] excludeFields) {
        return getDataById(index, id, null, excludeFields);
    }

    /**
     * 根据id获取详情
     *
     * @param index     索引名称
     * @param id        id
     * @param tClass
     * @return
     * @throws Exception
     */
    @Override
    public T getEntityById(String index, String id, Class<T> tClass) {
        Assert.isTrue(UtilString.isNotEmpty(id), "id不能为空");
        try {
            GetRequest request = new GetRequest(index, id);
            GetResponse response = restHighLevelClient.get(request, RequestOptions.DEFAULT);
            if (response.isExists()) {
                String dataStar = response.getSourceAsString();
                return JsonMapper.parseObject(dataStar, tClass);
            } else {
                return null;
            }
        } catch (Exception e) {
            log.error("获取详情信息失败：", e);
            throw new AppEsQueryException(e);
        }
    }

    /**
     * 根据查询方式获取某个索引下的文档个数
     *
     * @param index
     * @param queryBuilder
     * @return
     */
    @Override
    public long queryCount(String index, QueryBuilder queryBuilder) {
        try {
            ValueCountAggregationBuilder count = AggregationBuilders.count("count").field("id");
            SearchRequest request = new SearchRequest();
            request.indices(index);
            //构建查询
            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
            sourceBuilder.query(queryBuilder);
            sourceBuilder.aggregation(count);
            sourceBuilder.size(0);
            request.source(sourceBuilder);
            SearchResponse searchResponse = restHighLevelClient.search(request, RequestOptions.DEFAULT);
            Aggregations aggregations = searchResponse.getAggregations();
            ValueCount valueCount = (ValueCount) aggregations.asMap().get("count");
            return valueCount.getValue();
        } catch (Exception e) {
            log.error("获取统计数量失败", e);
            throw new AppEsQueryException(e);
        }
    }

    /**
     * 查看文档是否存在
     *
     * @param id
     * @return
     * @throws IOException
     */
    @Override
    public Boolean existsDocument(String index, String id) {
        Assert.isTrue(UtilString.isNotEmpty(id), "id不能为空");
        try {
            GetRequest getRequest = new GetRequest(index, id);
            return restHighLevelClient.exists(getRequest, RequestOptions.DEFAULT);
        } catch (Exception e) {
            log.error("查看文档是否失败异常", e);
            throw new AppEsQueryException(e);
        }
    }

    /**
     * 多条件组合查询
     * @param index         索引名称
     * @param queryBuilder  查询条件
     * @param resultClass   返回结果类型
     * @param size          查询数量
     * @return
     * @throws IOException
     */
    @Override
    public List<T> searchDocument(String index, QueryBuilder queryBuilder, Class<T> resultClass, int size) {
        try {
            SearchRequest request = new SearchRequest();
            request.indices(index);
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            if (queryBuilder != null) {
                searchSourceBuilder.query(queryBuilder);
            }
            searchSourceBuilder.size(size < 1 ? AppEsConstants.DEFAULT_UPDATE_BATCH_SIZE : size);
            request.source(searchSourceBuilder);
            SearchResponse searchResponse = restHighLevelClient.search(request, RequestOptions.DEFAULT);
            SearchHits hits1 = searchResponse.getHits();
            SearchHit[] hits2 = hits1.getHits();
            List<T> resultList = new ArrayList<>(hits2.length);
            for (SearchHit hit : hits2) {
                String strJson = hit.getSourceAsString();
                resultList.add(JsonMapper.parseObject(strJson, resultClass));
            }
            return resultList;
        } catch (Exception e) {
            log.error("查询用户信息失败", e);
            throw new AppEsQueryException(e);
        }
    }

    /**
     * scroll 查询
     * @param index             索引名
     * @param scroll            滚动查询有效时间设置
     * @param queryBuilder      查询信息
     * @param size              查询条数
     * @param sortBuilders      排序条件
     * @param resultClass       返回类型
     * @return
     */
    @Override
    public AppEsScrollPage<T> queryScrollPage(String index, Scroll scroll, QueryBuilder queryBuilder, int size, List<SortBuilder> sortBuilders, Class<T> resultClass) {
        try {
            SearchRequest request = new SearchRequest();
            request.indices(index);
            request.scroll(scroll);
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            searchSourceBuilder.query(queryBuilder);
            searchSourceBuilder.size(size < 1 ? AppEsConstants.DEFAULT_UPDATE_BATCH_SIZE : size);
            //排序字段
            if (UtilCollection.isNotEmpty(sortBuilders)) {
                for (SortBuilder sortBuilder : sortBuilders) {
                    searchSourceBuilder.sort(sortBuilder);
                }
            }
            request.source(searchSourceBuilder);
            // 执行搜索,返回搜索响应信息
            SearchResponse searchResponse = restHighLevelClient.search(request, RequestOptions.DEFAULT);
            TotalHits totalHits = searchResponse.getHits().getTotalHits();
            long length = searchResponse.getHits().getHits().length;
            if (searchResponse.status().getStatus() == STATUS_STATUS) {
                String scrollId = searchResponse.getScrollId();
                // 解析对象
                SearchHits hits1 = searchResponse.getHits();
                SearchHit[] hits2 = hits1.getHits();
                List<T> resultList = new ArrayList<T>();
                for (SearchHit hit : hits2) {
                    String strJson = hit.getSourceAsString();
                    resultList.add(JsonMapper.parseObject(strJson, resultClass));
                }
                if (hits2.length < size) {
                    clearScrollRequest(scrollId);
                }
                log.debug("共查询到[{}]条数据,处理数据条数[{}]", totalHits, length);
                return new AppEsScrollPage(scrollId, size, resultList);
            } else {
                log.error("scroll 查询失败：{}", searchResponse.toString());
                throw new AppEsQueryException(searchResponse.toString());
            }
        } catch (Exception e) {
            log.error("scroll 查询失败：", e);
            throw new AppEsQueryException(e);
        }
    }


    /**
     *
     * @param scroll
     * @param scrollId
     * @param size
     * @param resultClass
     * @return
     */
    @Override
    public AppEsScrollPage<T> queryScrollPage(Scroll scroll, String scrollId, int size, Class<T> resultClass) {
        SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
        scrollRequest.scroll(scroll);
        try {
            SearchResponse response = restHighLevelClient.scroll(scrollRequest, RequestOptions.DEFAULT);
            TotalHits totalHits = response.getHits().getTotalHits();
            long length = response.getHits().getHits().length;
            if (response.status().getStatus() == STATUS_STATUS) {
                scrollId = response.getScrollId();
                // 解析对象
                SearchHits hits1 = response.getHits();
                SearchHit[] hits2 = hits1.getHits();
                List<T> resultList = new ArrayList<T>();
                for (SearchHit hit : hits2) {
                    String strJson = hit.getSourceAsString();
                    resultList.add(JsonMapper.parseObject(strJson, resultClass));
                }
                if (hits2.length < size) {
                    clearScrollRequest(scrollId);
                }
                log.debug("共查询到[{}]条数据,处理数据条数[{}]", totalHits, length);
                return new AppEsScrollPage(scrollId, size, resultList);
            } else {
                log.error("scroll 查询失败：{}", response.toString());
                throw new AppEsQueryException(response.toString());
            }
        } catch (Exception e) {
            log.error("scroll 查询失败：", e);
            throw new AppEsQueryException(e);
        }
    }



    /**
     * 关闭scroll请求
     * @param scrollId
     * @return
     */
    private boolean clearScrollRequest(String scrollId) {
        try {
            ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
            //也可以选择setScrollIds()将多个scrollId一起使用
            clearScrollRequest.addScrollId(scrollId);
            ClearScrollResponse clearScrollResponse = restHighLevelClient.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
            return clearScrollResponse.isSucceeded();
        } catch (Exception e) {
            throw new AppEsUpdateException(e);
        }
    }

    /**
     * 统计查询
     *
     * @param index              索引
     * @param queryBuilder       查询条件
     * @param aggregationBuilder
     * @return
     * @throws Exception
     */
    @Override
    public Aggregations aggregation(String index, QueryBuilder queryBuilder, AggregationBuilder aggregationBuilder) {
        try {
            SearchRequest request = new SearchRequest();
            request.indices(index);
            //构建查询
            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
            sourceBuilder.query(queryBuilder);
            sourceBuilder.aggregation(aggregationBuilder);
            sourceBuilder.size(0);
            request.source(sourceBuilder);
            SearchResponse searchResponse = restHighLevelClient.search(request, RequestOptions.DEFAULT);
            if (searchResponse.status().getStatus() == STATUS_STATUS) {
                return searchResponse.getAggregations();
            } else {
                throw new AppEsQueryException(searchResponse.toString());
            }
        } catch (Exception e) {
            log.error("统计查询失败：", e);
            throw new AppEsQueryException(e);
        }
    }

    /**
     * 使用分词查询,并分页
     *
     * @param startPage     当前页         从0开始
     * @param pageSize      每页显示条数
     * @param sortBuilders  排序字段
     * @return
     */
    @Override
    public AppEsPage<T> queryPage(String index, int startPage, int pageSize, QueryBuilder queryBuilder, List<SortBuilder> sortBuilders, Class<T> resultClass) {
        try {
            startPage = startPage < 0 ? 0 : startPage;
            pageSize = pageSize < 0 ? AppEsConstants.DEFAULT_PAGE_SIZE : pageSize;

            SearchRequest request = new SearchRequest();
            request.indices(index);
            //构建搜索条件
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            //排序字段
            if (UtilCollection.isNotEmpty(sortBuilders)) {
                for (SortBuilder sortBuilder : sortBuilders) {
                    searchSourceBuilder.sort(sortBuilder);
                }
            }
            if (queryBuilder != null) {
                searchSourceBuilder.query(queryBuilder);
            }
            int start = startPage * pageSize;
            searchSourceBuilder.from(start).size(pageSize);

            // 设置是否按查询匹配度排序
            searchSourceBuilder.explain(true);

            log.debug("分页查询条件{}", searchSourceBuilder);

            request.source(searchSourceBuilder);
            // 执行搜索,返回搜索响应信息
            SearchResponse searchResponse = restHighLevelClient.search(request, RequestOptions.DEFAULT);


            TotalHits totalHits = searchResponse.getHits().getTotalHits();
            long length = searchResponse.getHits().getHits().length;
            if (searchResponse.status().getStatus() == STATUS_STATUS) {
                // 解析对象
                SearchHits hits1 = searchResponse.getHits();
                SearchHit[] hits2 = hits1.getHits();
                List<T> resultList = new ArrayList<T>();
                for (SearchHit hit : hits2) {
                    String strJson = hit.getSourceAsString();
                    resultList.add(JsonMapper.parseObject(strJson, resultClass));
                }

                log.debug("共查询到[{}]条数据,处理数据条数[{}]", totalHits, length);

                return new AppEsPage(startPage, pageSize, (int) totalHits.value, resultList);
            } else {
                log.error("es分页查询失败：{}", searchResponse.toString());
                throw new AppEsQueryException("es分页查询失败");
            }
        } catch (Exception e) {
            log.error("分页查询失败：", e);
            throw new AppEsQueryException(e);
        }
    }

    /**
     * 使用分词查询,并分页
     *
     * @param startPage     当前页            从0开始分页
     * @param pageSize      每页显示条数
     * @param sortBuilders  排序字段
     * @return
     */
    @Override
    public AppEsPage<Map> queryPage(String index, int startPage, int pageSize, QueryBuilder queryBuilder, List<SortBuilder> sortBuilders, String[] includeFields) {
        try {
            startPage = startPage < 0 ? 0 : startPage;
            pageSize = pageSize < 0 ? AppEsConstants.DEFAULT_PAGE_SIZE : pageSize;
            SearchRequest request = new SearchRequest();
            request.indices(index);
            //构建搜索条件
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            if (includeFields != null && includeFields.length > 0) {
                searchSourceBuilder.fetchSource(includeFields, Strings.EMPTY_ARRAY);
            }
            //排序字段
            if (UtilCollection.isNotEmpty(sortBuilders)) {
                for (SortBuilder sortBuilder : sortBuilders) {
                    searchSourceBuilder.sort(sortBuilder);
                }
            }
            searchSourceBuilder.query(queryBuilder);
            int start = startPage * pageSize;
            searchSourceBuilder.from(start).size(pageSize);
            // 设置是否按查询匹配度排序
            searchSourceBuilder.explain(true);

            log.debug("分页查询条件{}", searchSourceBuilder);

            request.source(searchSourceBuilder);
            // 执行搜索,返回搜索响应信息
            SearchResponse searchResponse = restHighLevelClient.search(request, RequestOptions.DEFAULT);


            TotalHits totalHits = searchResponse.getHits().getTotalHits();
            long length = searchResponse.getHits().getHits().length;
            if (searchResponse.status().getStatus() == STATUS_STATUS) {
                // 解析对象
                SearchHits hits1 = searchResponse.getHits();
                SearchHit[] hits2 = hits1.getHits();
                List<Map> resultList = Lists.newArrayListWithCapacity(hits2.length);
                for (SearchHit hit : hits2) {
                    String strJson = hit.getSourceAsString();
                    resultList.add(JsonMapper.parseObject(strJson, Map.class));
                }

                log.debug("共查询到[{}]条数据,处理数据条数[{}]", totalHits, length);

                return new AppEsPage(startPage, pageSize, (int) totalHits.value, resultList);
            } else {
                log.error("es分页查询失败：{}", searchResponse.toString());
                throw new AppEsQueryException("es分页查询失败");
            }
        } catch (IOException e) {
            log.error("es分页查询失败", e);
            throw new AppEsQueryException(e);
        }
    }

    /**
     * 删除单条数据
     */
    @Override
    public void deleteDocument(String index,  String id) {
        Assert.isTrue(UtilString.isNotEmpty(id), "id不能为空");
        try {
            Assert.isTrue(UtilString.isNotEmpty(id), "id不能为空");
            DeleteRequest deleteRequest = new DeleteRequest(index, id);
            DeleteResponse delete = restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
            if (delete.status().getStatus() != STATUS_STATUS) {
                log.error("删除数据失败：{}", delete.toString());
                throw new AppEsDeleteException(delete.toString());
            }
        } catch (Exception e) {
            log.error("删除document失败", e);
            throw new AppEsDeleteException(e);
        }
    }

    /**
     * 批量删除数据
     *
     * @param ids
     * @return
     * @throws IOException
     */
    @Override
    public void deleteBatchDocument(String index, List<String> ids) {
        Assert.isTrue(UtilCollection.isNotEmpty(ids), "id不能为空");
        try {
            //创建请求对象indices()
            BulkRequest bulkRequest = new BulkRequest();
            ids.forEach(id -> bulkRequest.add(new DeleteRequest(index, id)));
            BulkResponse bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (bulkResponse.hasFailures()) {
                log.error("删除数据失败：{}", bulkResponse.buildFailureMessage());
                throw new AppEsDeleteException(bulkResponse.buildFailureMessage());
            }
        } catch (Exception e) {
            log.error("批量删除doc失败", e);
            throw new AppEsDeleteException(e);
        }

    }

    /**
     * 公共批量bulk处理数据 插入、更新、删除
     */
    @Override
    public void bulkBatchDocument(BulkRequest bulkRequest) {
        Assert.isTrue(UtilCollection.isNotEmpty(bulkRequest.requests()), "数据不能为空");
        try {
            BulkResponse response = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (response.hasFailures()) {
                log.error("批量处理数据失败：{}", response.buildFailureMessage());
                throw new AppEsCreateException(response.buildFailureMessage());
            }
        } catch (Exception e) {
            log.error("批量处理数据失败", e);
            throw new AppEsCreateException(e);
        }
    }
}
